#include "MonitorPid.hpp"
#include <errno.h>
#include <unistd.h>
#ifndef _WIN32
#include <sys/socket.h>
#include <linux/netlink.h>
#include <linux/connector.h>
#include <linux/cn_proc.h>
#include <fcntl.h>
#include <sys/resource.h>
#include <pthread.h>
#endif
#include <map>
#include <cstring>
#include <csignal>
#include <algorithm>
#include <iomanip>

namespace evbs {
namespace ertps {
namespace rtps {

// Constructor and destructor
ProcessExitMonitor::ProcessExitMonitor() : nl_sock_(-1), running_(false) {}

ProcessExitMonitor::~ProcessExitMonitor() {
    stop();
}

// Get signal names mapping
const std::map<int, std::string>& ProcessExitMonitor::get_signal_names() {
#ifndef _WIN32
    static const std::map<int, std::string> signal_names = {
        {SIGHUP, "SIGHUP"},   {SIGINT, "SIGINT"},       {SIGQUIT, "SIGQUIT"}, {SIGILL, "SIGILL"},
        {SIGTRAP, "SIGTRAP"}, {SIGABRT, "SIGABRT"},     {SIGBUS, "SIGBUS"},   {SIGFPE, "SIGFPE"},
        {SIGKILL, "SIGKILL"}, {SIGUSR1, "SIGUSR1"},     {SIGSEGV, "SIGSEGV"}, {SIGUSR2, "SIGUSR2"},
        {SIGPIPE, "SIGPIPE"}, {SIGALRM, "SIGALRM"},     {SIGTERM, "SIGTERM"}, {SIGSTKFLT, "SIGSTKFLT"},
        {SIGCHLD, "SIGCHLD"}, {SIGCONT, "SIGCONT"},     {SIGSTOP, "SIGSTOP"}, {SIGTSTP, "SIGTSTP"},
        {SIGTTIN, "SIGTTIN"}, {SIGTTOU, "SIGTTOU"},     {SIGURG, "SIGURG"},   {SIGXCPU, "SIGXCPU"},
        {SIGXFSZ, "SIGXFSZ"}, {SIGVTALRM, "SIGVTALRM"}, {SIGPROF, "SIGPROF"}, {SIGWINCH, "SIGWINCH"},
        {SIGIO, "SIGIO"},     {SIGPWR, "SIGPWR"},       {SIGSYS, "SIGSYS"}};
    return signal_names;
#else
    elogWarning(RTPS_PDP_SERVER, "ProcessExitMonitor::get_signal_names() not supported on Windows");
    return std::map<int, std::string>();
#endif
}

void ProcessExitMonitor::dump_monitored_pids() const {
    std::lock_guard<std::mutex> lock(pids_mutex_);
    logInfo(RTPS_PDP_SERVER, "=== Monitored PIDs Dump ===");
    logInfo(RTPS_PDP_SERVER, "Total monitored PIDs: " << monitored_pids_.size());
    for (const auto& entry : monitored_pids_) {
        int pid = entry.first;
        const auto& guids = entry.second;
        std::string proc_name = get_process_name(pid);
        logInfo(RTPS_PDP_SERVER, "PID " << pid << " (" << proc_name << ") - GUIDs: " << guids.size());
        for (const auto& guid : guids) {
            logInfo(RTPS_PDP_SERVER, "  - " << guid);
        }
    }
    logInfo(RTPS_PDP_SERVER, "==========================");
}

void ProcessExitMonitor::set_exit_callback(std::function<void(int pid, const GUID_t& guid)> callback) {
    exit_callback_ = callback;
}

bool ProcessExitMonitor::init() {
#ifndef _WIN32
    // Set process priority
    if (setpriority(PRIO_PROCESS, 0, -10) == 0) {
        logInfo(RTPS_PDP_SERVER, "Process priority set to -10 (high priority)");
    } else {
        elogWarning(RTPS_PDP_SERVER, "Failed to set process priority: " << strerror(errno));
    }
    nl_sock_ = socket(PF_NETLINK, SOCK_DGRAM, NETLINK_CONNECTOR);
    if (nl_sock_ < 0) {
        elogError(RTPS_PDP_SERVER, RetCode_t::RETCODE_BAD_PARAMETER, "monitorPid socket error: " << strerror(errno));
        return false;
    }

    // Set receive buffer size
    int rcvbuf_size = 16 * 1024 * 1024;
    if (setsockopt(nl_sock_, SOL_SOCKET, SO_RCVBUF, &rcvbuf_size, sizeof(rcvbuf_size)) < 0) {
        elogError(RTPS_PDP_SERVER, RetCode_t::RETCODE_BAD_PARAMETER,
                  "monitorPid setsockopt SO_RCVBUF error: " << strerror(errno));
    }

    // Verify the actual buffer size
    int actual_rcvbuf_size;
    socklen_t optlen = sizeof(actual_rcvbuf_size);
    if (getsockopt(nl_sock_, SOL_SOCKET, SO_RCVBUF, &actual_rcvbuf_size, &optlen) == 0) {
        logInfo(RTPS_PDP_SERVER,
                "Requested socket receive buffer size: " << rcvbuf_size << ", Actual size: " << actual_rcvbuf_size);
        if (actual_rcvbuf_size < rcvbuf_size) {
            elogWarning(RTPS_PDP_SERVER, "Actual buffer size is smaller than requested, may cause drops");
        }
    }

    // Check system-level limits
    FILE* file = fopen("/proc/sys/net/core/rmem_max", "r");
    if (file) {
        char line[64];
        if (fgets(line, sizeof(line), file)) {
            int sys_rmem_max = atoi(line);
            logInfo(RTPS_PDP_SERVER, "System max receive buffer size: " << sys_rmem_max);
            if (sys_rmem_max < rcvbuf_size) {
                elogWarning(RTPS_PDP_SERVER, "System limit is smaller than requested buffer size");
            }
        }
        fclose(file);
    }

    // Set socket timeout
    struct timeval timeout;
    timeout.tv_sec = 10;
    timeout.tv_usec = 0;  // 10 second timeout
    if (setsockopt(nl_sock_, SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof(timeout)) < 0) {
        elogError(RTPS_PDP_SERVER, RetCode_t::RETCODE_BAD_PARAMETER,
                  "monitorPid setsockopt SO_RCVTIMEO error: " << strerror(errno));
    }

    struct sockaddr_nl sa;
    memset(&sa, 0, sizeof(sa));
    sa.nl_family = AF_NETLINK;
    sa.nl_groups = CN_IDX_PROC;
    sa.nl_pid = getpid();

    if (bind(nl_sock_, (struct sockaddr*)&sa, sizeof(sa)) < 0) {
        elogError(RTPS_PDP_SERVER, RetCode_t::RETCODE_BAD_PARAMETER, "monitorPid bind error: " << strerror(errno));
        close(nl_sock_);
        return false;
    }

    // Subscribe to proc events
    constexpr std::size_t req_size(sizeof(nlmsghdr) + sizeof(cn_msg) + sizeof(enum proc_cn_mcast_op));
    std::vector<uint8_t> nl_req;
    nl_req.resize(req_size);

    // nl_req contains nlmsg header, cn_msg and op types
    nlmsghdr *nl_hdr = reinterpret_cast<struct nlmsghdr *>(nl_req.data());
    cn_msg *cn_msg = reinterpret_cast<struct cn_msg *>(nl_hdr+1);
    enum proc_cn_mcast_op *op = reinterpret_cast<enum proc_cn_mcast_op *>(cn_msg+1);

    nl_hdr->nlmsg_len = req_size;
    nl_hdr->nlmsg_type = NLMSG_DONE;
    nl_hdr->nlmsg_flags = 0;
    nl_hdr->nlmsg_pid = getpid();
    nl_hdr->nlmsg_seq = 0;
    cn_msg->id.idx = CN_IDX_PROC;
    cn_msg->id.val = CN_VAL_PROC;
    cn_msg->len = sizeof(enum proc_cn_mcast_op);
    *op = PROC_CN_MCAST_LISTEN;

    if (send(nl_sock_, &nl_req, req_size, 0) < 0) {
        elogError(RTPS_PDP_SERVER, RetCode_t::RETCODE_BAD_PARAMETER, "monitorPid send error: " << strerror(errno));
        close(nl_sock_);
        std::vector<uint8_t>().swap(nl_req);
        return false;
    }

    std::vector<uint8_t>().swap(nl_req);

    logDebug(RTPS_PDP_SERVER, "ProcessExitMonitor initialized successfully");
    return true;
#else
    logWarning(RTPS_PDP_SERVER, "Process monitoring not supported on Windows");
    return false;
#endif
}

void ProcessExitMonitor::start_monitoring() {
#ifndef _WIN32
    if (running_) {
        logWarning(RTPS_PDP_SERVER, "ProcessExitMonitor is already running");
        return;
    }

    running_ = true;
    signal(SIGUSR1, signal_handler);
    monitor_thread_ = std::thread(&ProcessExitMonitor::run, this);

    // set monitor thread priority
    sched_param sch_params;
    sch_params.sched_priority = 50;  // real-time priority 50
    if (pthread_setschedparam(monitor_thread_.native_handle(), SCHED_FIFO, &sch_params) == 0) {
        logInfo(RTPS_PDP_SERVER, "Monitor thread priority set to SCHED_FIFO:50");
    } else {
        elogWarning(RTPS_PDP_SERVER, "Failed to set thread priority: " << strerror(errno));
    }

    // Set thread name
    std::string thread_name = "ExitMonitor";
    int ret = pthread_setname_np(monitor_thread_.native_handle(), thread_name.c_str());
    if (ret == 0) {
        logInfo(RTPS_PDP_SERVER, "Monitor thread name set to: " << thread_name);
    } else {
        elogWarning(RTPS_PDP_SERVER, "Failed to set thread name: " << ret << " " << thread_name);
    }
#endif

    logDebug(RTPS_PDP_SERVER, "ProcessExitMonitor started");
}

void ProcessExitMonitor::stop() {
#ifndef _WIN32
    if (!running_) {
        return;
    }

    running_ = false;

    if (monitor_thread_.joinable()) {
        pthread_kill(monitor_thread_.native_handle(), SIGUSR1);
        monitor_thread_.join();
    }

    if (nl_sock_ >= 0) {
        shutdown(nl_sock_, SHUT_RDWR);
        close(nl_sock_);
        nl_sock_ = -1;
    }

    logDebug(RTPS_PDP_SERVER, "ProcessExitMonitor stopped");
#else
    logWarning(RTPS_PDP_SERVER, "ProcessExitMonitor not supported on Windows");
#endif
}

void ProcessExitMonitor::add_pid(int pid, const GUID_t& guid) {
    std::lock_guard<std::mutex> lock(pids_mutex_);
    if (guid == GUID_t::unknown()) {
        // If no GUID is provided, only add PID (backward compatibility)
        if (monitored_pids_.find(pid) == monitored_pids_.end()) {
            monitored_pids_[pid] = std::vector<GUID_t>();
        }
    } else {
        // Add PID and GUID mapping
        monitored_pids_[pid].push_back(guid);
    }
    logInfo(RTPS_PDP_SERVER, "Added PID " << pid << " for monitoring");
}

void ProcessExitMonitor::remove_pid_guid(int pid, const GUID_t& guid) {
    std::lock_guard<std::mutex> lock(pids_mutex_);
    auto it = monitored_pids_.find(pid);
    if (it != monitored_pids_.end()) {
        auto& guid_list = it->second;
        guid_list.erase(std::remove(guid_list.begin(), guid_list.end(), guid), guid_list.end());
        // If the PID has no more GUIDs, remove the entire PID entry
        if (guid_list.empty()) {
            monitored_pids_.erase(it);
            logInfo(RTPS_PDP_SERVER, "Removed PID " << pid << " from monitoring (no more GUIDs)");
        } else {
            logInfo(RTPS_PDP_SERVER, "Removed GUID from PID " << pid << ", remaining GUIDs: " << guid_list.size());
        }
    }
}

void ProcessExitMonitor::remove_pid(int pid) {
    std::lock_guard<std::mutex> lock(pids_mutex_);
    auto it = monitored_pids_.find(pid);
    if (it != monitored_pids_.end()) {
        monitored_pids_.erase(it);
        logDebug(RTPS_PDP_SERVER, "Removed PID " << pid << " from monitoring");
    }
}

std::string ProcessExitMonitor::get_signal_name(int signal_num) {
    const auto& signal_names = get_signal_names();
    auto it = signal_names.find(signal_num);
    if (it != signal_names.end()) {
        return it->second;
    }
    return "SIGUNKNOWN(" + std::to_string(signal_num) + ")";
}

std::string ProcessExitMonitor::analyze_exit_reason(int exit_code, int exit_signal) {
#ifndef _WIN32
    if (exit_signal > 0) {
        std::string signal_name = get_signal_name(exit_signal);
        std::string reason;

        switch (exit_signal) {
            case SIGTERM:
                reason = "Terminated normally";
                break;
            case SIGKILL:
                reason = "Forcefully killed";
                break;
            case SIGSEGV:
                reason = "Segmentation fault";
                break;
            case SIGABRT:
                reason = "Aborted (assertion failure or abort() call)";
                break;
            case SIGBUS:
                reason = "Bus error";
                break;
            case SIGFPE:
                reason = "Floating point exception";
                break;
            case SIGILL:
                reason = "Illegal instruction";
                break;
            case SIGPIPE:
                reason = "Broken pipe";
                break;
            case SIGALRM:
                reason = "Timeout (alarm)";
                break;
            case SIGUSR1:
            case SIGUSR2:
                reason = "User-defined signal";
                break;
            default:
                reason = "Killed by signal";
                break;
        }
        return reason + " (" + signal_name + ")";
    } else if (exit_code != 0) {
        return "Exited with error code " + std::to_string(exit_code);
    } else {
        return "Exited normally";
    }
#else
    return "Windows not support";
#endif
}

void ProcessExitMonitor::run() {
#ifndef _WIN32
    char buf[262144];
    auto last_report_time = std::chrono::steady_clock::now();

    // reset statistics variables
    total_events_ = 0;
    dropped_events_ = 0;
    monitored_exits_ = 0;
    non_monitored_exits_ = 0;
    {
        std::lock_guard<std::mutex> lock(stats_mutex_);
        non_monitored_processes_.clear();
        non_monitored_exit_codes_.clear();
        non_monitored_signals_.clear();
    }

    while (running_) {
        int len = recv(nl_sock_, buf, sizeof(buf), 0);

        if (len < 0) {
            if (errno == EINTR) {
                continue;
            } else if (errno == EAGAIN || errno == EWOULDBLOCK) {
                // This should not happen in blocking mode, but handle for safety
                logDebug(RTPS_PDP_SERVER, "Unexpected EAGAIN/EWOULDBLOCK in blocking mode");
                continue;
            } else if (errno == ENOBUFS) {
                dropped_events_++;
                elogWarning(RTPS_PDP_SERVER,
                            "Netlink buffer overflow detected, some events may be lost (total dropped: "
                                << dropped_events_.load() << ")");
                std::this_thread::sleep_for(std::chrono::milliseconds(1));  // 10ms
                continue;
            } else {
                elogWarning(RTPS_PDP_SERVER, "monitorPid recv error: " << strerror(errno));
                std::this_thread::sleep_for(std::chrono::milliseconds(10));  // 10ms
                continue;
            }
        }

        struct nlmsghdr* nlh = (struct nlmsghdr*)buf;

        // get all exit events
        for (; NLMSG_OK(nlh, len); nlh = NLMSG_NEXT(nlh, len)) {
            if (nlh->nlmsg_type == NLMSG_NOOP || nlh->nlmsg_type == NLMSG_ERROR) {
                continue;
            }

            struct cn_msg* cn_hdr = (struct cn_msg*)NLMSG_DATA(nlh);
            if (cn_hdr->id.idx != CN_IDX_PROC || cn_hdr->id.val != CN_VAL_PROC) {
                continue;
            }

            struct proc_event* ev = (struct proc_event*)cn_hdr->data;
            if (ev->what == proc_event::PROC_EVENT_EXIT) {
                int pid = ev->event_data.exit.process_pid;
                int tgid = ev->event_data.exit.process_tgid;
                int exit_code = ev->event_data.exit.exit_code;
                int exit_signal = ev->event_data.exit.exit_signal;
                // quick filter: only process we care about
                bool is_monitored = false;
                {
                    std::lock_guard<std::mutex> lock(pids_mutex_);
                    is_monitored = (monitored_pids_.find(pid) != monitored_pids_.end());
                }

                if (!is_monitored) {
                    // not monitored process, record statistics and skip
                    non_monitored_exits_++;

                    if (is_enable_detailed_logging()) {
                        // record detailed information of non-monitored processes
                        std::string proc_name = get_process_name(pid);
                        {
                            std::lock_guard<std::mutex> lock(stats_mutex_);
                            non_monitored_processes_[proc_name]++;
                            non_monitored_exit_codes_[exit_code]++;
                            non_monitored_signals_[exit_signal]++;
                        }
                    }

                    continue;
                } else {
                    logDebug(RTPS_PDP_SERVER, "Processing monitored process exit - PID: " << pid);
                }

                // only record monitored process exit events, reduce log volume
                logInfo(RTPS_PDP_SERVER, "Monitored process exit - PID: " << pid << ", TGID: " << tgid
                                                                          << ", Exit code: " << exit_code
                                                                          << ", Signal: " << exit_signal);
                std::vector<GUID_t> affected_guids;
                {
                    std::lock_guard<std::mutex> lock(pids_mutex_);
                    auto it = monitored_pids_.find(pid);
                    if (it != monitored_pids_.end()) {
                        affected_guids = it->second;
                        monitored_pids_.erase(it);
                    }
                }
                if (!affected_guids.empty()) {
                    monitored_exits_++;

                    // record detailed process exit information
                    std::string exit_reason = analyze_exit_reason(exit_code, exit_signal);
                    logInfo(RTPS_PDP_SERVER, "=== Process Exit Detection ===");
                    logInfo(RTPS_PDP_SERVER, "PID: " << pid << " (TGID: " << tgid << ")");
                    logInfo(RTPS_PDP_SERVER, "Exit Code: " << exit_code);
                    logInfo(RTPS_PDP_SERVER, "Exit Signal: " << exit_signal);
                    logInfo(RTPS_PDP_SERVER, "Reason: " << exit_reason);
                    logInfo(RTPS_PDP_SERVER, "Affected GUIDs (" << affected_guids.size() << "):");
                    for (const auto& guid : affected_guids) {
                        logInfo(RTPS_PDP_SERVER, "  - " << guid);
                        // call callback function
                        if (exit_callback_) {
                            try {
                                exit_callback_(pid, guid);
                            } catch (const std::exception& e) {
                                elogError(RTPS_PDP_SERVER, RetCode_t::RETCODE_ERROR,
                                          "Exception in exit callback: " << e.what());
                            }
                        }
                    }
                    logInfo(RTPS_PDP_SERVER, "GUIDs added to deletion list: " << affected_guids.size());
                    logInfo(RTPS_PDP_SERVER, "================================");

                } else {
                    // this should not happen, because we have confirmed that the process is in the monitoring list
                    logWarning(RTPS_PDP_SERVER, "Unexpected: monitored process " << pid << " has no GUIDs");
                }
            }
        }

        // Report statistics every 100 seconds
        auto now = std::chrono::steady_clock::now();
        if ((now - last_report_time > std::chrono::seconds(100) || is_enable_detailed_logging()) &&
            dropped_events_.load() > 0) {
            double filter_rate =
                (total_events_.load() + non_monitored_exits_.load()) > 0
                    ? (double)non_monitored_exits_.load() / (total_events_.load() + non_monitored_exits_.load()) * 100.0
                    : 0.0;

            logInfo(RTPS_PDP_SERVER, "ProcessExitMonitor stats - Total events: "
                                         << total_events_.load() << ", Monitored exits: " << monitored_exits_.load()
                                         << ", Non-monitored exits: " << non_monitored_exits_.load()
                                         << ", Filter rate: " << std::fixed << std::setprecision(1) << filter_rate
                                         << "%"
                                         << ", Dropped events: " << dropped_events_.load()
                                         << ", Processing rate: " << (monitored_exits_.load() / 30) << " events/sec");

            // output current monitored process information
            dump_monitored_pids();

            // output detailed statistics of non-monitored processes
            if (is_enable_detailed_logging()) {
                std::lock_guard<std::mutex> lock(stats_mutex_);
                logInfo(RTPS_PDP_SERVER, "Debug: non_monitored_processes_ size: " << non_monitored_processes_.size());
                logInfo(RTPS_PDP_SERVER, "Debug: non_monitored_exit_codes_ size: " << non_monitored_exit_codes_.size());
                logInfo(RTPS_PDP_SERVER, "Debug: non_monitored_signals_ size: " << non_monitored_signals_.size());

                // print all collected process names (for debugging)
                if (!non_monitored_processes_.empty()) {
                    logInfo(RTPS_PDP_SERVER, "Debug: All collected process names:");
                    for (const auto& pair : non_monitored_processes_) {
                        logInfo(RTPS_PDP_SERVER, "  " << pair.first << ": " << pair.second);
                    }
                }

                if (!non_monitored_processes_.empty()) {
                    logInfo(RTPS_PDP_SERVER, "=== Non-monitored processes summary ===");

                    // by process name
                    logInfo(RTPS_PDP_SERVER, "Top non-monitored processes:");
                    std::vector<std::pair<std::string, int>> sorted_processes;
                    for (const auto& pair : non_monitored_processes_) {
                        sorted_processes.emplace_back(pair.first, pair.second);
                    }
                    std::sort(sorted_processes.begin(), sorted_processes.end(),
                              [](const auto& a, const auto& b) { return a.second > b.second; });

                    for (size_t i = 0; i < std::min(size_t(5), sorted_processes.size()); i++) {
                        logInfo(RTPS_PDP_SERVER,
                                "  " << sorted_processes[i].first << ": " << sorted_processes[i].second << " exits");
                    }

                    // by exit code
                    logInfo(RTPS_PDP_SERVER, "Top exit codes:");
                    std::vector<std::pair<int, int>> sorted_codes;
                    for (const auto& pair : non_monitored_exit_codes_) {
                        sorted_codes.emplace_back(pair.first, pair.second);
                    }
                    std::sort(sorted_codes.begin(), sorted_codes.end(),
                              [](const auto& a, const auto& b) { return a.second > b.second; });

                    for (size_t i = 0; i < std::min(size_t(3), sorted_codes.size()); i++) {
                        logInfo(RTPS_PDP_SERVER,
                                "  Exit code " << sorted_codes[i].first << ": " << sorted_codes[i].second << " times");
                    }

                    // by signal
                    logInfo(RTPS_PDP_SERVER, "Top signals:");
                    std::vector<std::pair<int, int>> sorted_signals;
                    for (const auto& pair : non_monitored_signals_) {
                        sorted_signals.emplace_back(pair.first, pair.second);
                    }
                    std::sort(sorted_signals.begin(), sorted_signals.end(),
                              [](const auto& a, const auto& b) { return a.second > b.second; });

                    for (size_t i = 0; i < std::min(size_t(3), sorted_signals.size()); i++) {
                        std::string signal_name = get_signal_name(sorted_signals[i].first);
                        logInfo(RTPS_PDP_SERVER, "  Signal " << sorted_signals[i].first << " (" << signal_name
                                                             << "): " << sorted_signals[i].second << " times");
                    }

                    logInfo(RTPS_PDP_SERVER, "=====================================");
                }
            }
            last_report_time = now;
        }
    }
#else
    logWarning(RTPS_PDP_SERVER, "Process monitoring not supported on Windows");
#endif
}

std::string ProcessExitMonitor::get_process_name(int pid) const {
#ifndef _WIN32
    std::string proc_path = "/proc/" + std::to_string(pid) + "/comm";
    FILE* file = fopen(proc_path.c_str(), "r");
    if (file) {
        char name[256];
        if (fgets(name, sizeof(name), file)) {
            size_t len = strlen(name);
            if (len > 0 && name[len - 1] == '\n') {
                name[len - 1] = '\0';
            }
            fclose(file);
            return std::string(name);
        }
        fclose(file);
    }
#endif
    return "unknown";
}

void ProcessExitMonitor::enable_detailed_logging(bool enable) {
    enable_detailed_logging_ = enable;
}
bool ProcessExitMonitor::is_enable_detailed_logging() const {
    return enable_detailed_logging_;
}

}  // namespace rtps
}  // namespace ertps
}  // namespace evbs